package com.hisense.ovcloud.mqtt.router.service;

import com.alibaba.fastjson2.JSON;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttDeviceUploadMsg;
import com.hisense.ovcloud.mqtt.router.config.Constants;
import com.hisense.ovcloud.mqtt.router.config.RouterConfig;
import io.micrometer.common.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
@Slf4j
@Lazy
public class ReactiveConsumerService implements CommandLineRunner {

    private final ThirdPartyApi thirdPartyApi;

    private final RouterConfig routerConfig;

    final
    ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate;

    public ReactiveConsumerService(ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate, RouterConfig routerConfig, ThirdPartyApi thirdPartyApi) {
        this.kafkaConsumerTemplate = kafkaConsumerTemplate;
        this.routerConfig = routerConfig;
        this.thirdPartyApi = thirdPartyApi;
    }

    private Flux<Mono<Boolean>> msgConsummer() {
        return kafkaConsumerTemplate
                .receiveAutoAck()
                .map(this::handler);
    }


    protected Mono<Boolean> handler(ConsumerRecord<String, String> consumerRecord) {
        try {
            if (StringUtils.isNotEmpty(consumerRecord.value())) {
                String receivedTxt = consumerRecord.value().strip().replaceAll("\n", "");
                if (consumerRecord.topic().equals(routerConfig.getUploadKafkaTopic())) {
                    MqttDeviceUploadMsg uploadMsg = JSON.parseObject(receivedTxt, MqttDeviceUploadMsg.class);
                    if(uploadMsg.getType().equals(Constants.MQTT.UploadMsgType.TT) || uploadMsg.getType().equals(Constants.MQTT.UploadMsgType.All)){
                        log.info("will send upload transparent transfer type upload msg to hitachi: {} ", receivedTxt);
                        return thirdPartyApi.notifyThirdPartyApi(JSON.parseObject(receivedTxt, MqttDeviceUploadMsg.class));
                    }
                }
                return Mono.just(true);
            }
        } catch (Exception e) {
            log.error("consuming encountered error: {}", e.getMessage());
        }
        return Mono.just(true);
    }

    @Override
    public void run(String... args) {
        msgConsummer().subscribe(Mono::subscribe);
    }
}

